import json
import uuid
import asyncio
from typing import Union
from functools import partial
from alibabacloud_green20220302.client import Client
from alibabacloud_green20220302 import models
from alibabacloud_tea_openapi.models import Config
from alibabacloud_tea_util import models as util_models
from common.utils.config import ConfigUtil


class AliyunCensor:
    """
    https://www.aliyun.com/product/lvwang
    """

    @classmethod
    async def client(cls, configs: dict = None):
        """
        构建客户端

        Args:
            configs (dict): 配置参数

        Author:
            zero
        """
        if configs is None:
            configs = await ConfigUtil.get("content_censor", "aliyun", {})

        if not configs.get("access_key_id"):
            raise Exception("阿里内容审核: 请在后台配置 access_key_id")

        if not configs.get("access_key_secret"):
            raise Exception("阿里内容审核: 请在后台配置 access_key_secret")

        server_config = Config(
            access_key_id=configs.get("access_key_id", ""),
            access_key_secret=configs.get("access_key_secret", ""),
            connect_timeout=10000,
            read_timeout=3000,
            region_id="cn-shanghai",
            endpoint="green-cip.cn-shanghai.aliyuncs.com"
        )

        return Client(server_config)

    @classmethod
    async def text(cls, text: str, service: str = "comment_detection_pro", version: str = "plus") -> Union[str, None]:
        """
        文本审核
        PS: 默认使用Plus版本, 因为普通版有内容长度限制。
        流量限制: 本接口的单用户QPS限制为100次/秒, 超过限制API调用会被限流。
        内容限制: 审核的文本内容, 限定在600字符以内 (通用型)。

        Args:
            text (str): 待审核的文本
            service (str): 审核服务类型
            version (str): 服务版本 (general=通用, plus=高级)

        Returns:
            str | None: str=不通过原因, None=通过

        Author:
            zero
        """
        if not text:
            return None

        if version == "general":
            return await cls.text_general(text, service)
        elif version == "plus":
            return await cls.text_plus(text, service)
        else:
            return f"Error: Aliyun content_censor not supported {version}"

    @classmethod
    async def text_general(cls, text: str, service: str = "comment_detection") -> Union[str, None]:
        """
        文本审核增强版 (通用版)
        计费单价: 7.5元/万次
        内容限制: 审核的文本内容, 限定在600字符以内。

        Args:
            text (str): 待审核的文本 (600字符以内)
            service (str): 审核服务类型
                - nickname_detection: 用户昵称检测
                - chat_detection: 私聊互动内容检测
                - comment_detection: 公聊评论内容检测
                - ai_art_detection: AIGC文字检测
                - ad_compliance_detection: 广告法合规检测
                - pgc_detection: PGC教学物料检测

        Returns:
            str | None: str=不通过原因, None=通过

        Author:
            zero
        """
        if not text:
            return None

        text_moderation_request = models.TextModerationRequest(
            service=service,
            service_parameters=json.dumps({
                "content": text,
                "dataId": str(uuid.uuid1())
            })
        )

        runtime = util_models.RuntimeOptions()
        runtime.read_timeout = 10000
        runtime.connect_timeout = 10000
        try:
            client = await cls.client()
            loop = asyncio.get_event_loop()
            func = partial(client.text_moderation_with_options, text_moderation_request, runtime)
            response = await loop.run_in_executor(None, func)
            
            if response.status_code == 200:
                result = response.body
                status = cls._handle_error(result.code)
                if status == "200":
                    resultData = result.data
                    reason = json.loads(resultData.reason)
                    if reason["riskLevel"] not in ["none", "low"]:
                        return f"{resultData.descriptions} - {reason['riskWords']}"
                    return None
                return status
            else:
                return None
        except Exception as err:
            return f"{str(err)} - (AliyunCensor)"

    @classmethod
    async def text_plus(cls, text: str, service: str = "comment_detection_pro") -> Union[str, None]:
        """
        文本审核增强版 (Plus)
        计费单价: 7.5元/万次

        Args:
            text (str): 待审核的文本
            service (str): 审核服务类型
                - nickname_detection_pro: 用户昵称检测
                - chat_detection_pro: 私聊互动内容检测
                - comment_detection_pro: 公聊评论内容检测
                - ad_compliance_detection_pro: 广告法合规检测

        Returns:
            str | None: str=不通过原因, None=通过

        Author:
            zero
        """
        if not text:
            return None

        text_moderation_request = models.TextModerationPlusRequest(
            service=service,
            service_parameters=json.dumps({
                "content": text
            })
        )

        try:
            client = await cls.client()
            loop = asyncio.get_event_loop()
            func = partial(client.text_moderation_plus, text_moderation_request)
            response = await loop.run_in_executor(None, func)
            
            if response.status_code == 200:
                result = response.body
                status = cls._handle_error(result.code)
                if status == "200":
                    resultData = result.data
                    if resultData.risk_level not in ["none", "low"]:
                        description = resultData.result[0].description
                        risk_words = resultData.result[0].risk_words
                        return f"{description} - {risk_words}"
                    return None
                return status
            else:
                return None
        except Exception as err:
            return f"{str(err)} - (AliyunCensor)"

    @classmethod
    async def image(cls, path: str, service: str = "baselineCheck") -> Union[str, None]:
        """
        图片审核通用 (image_standard)

        Args:
            path (str): 图片地址,如: https://www.api.com/aa.png
            service (str): 业务场景 15元/万次
                - baselineCheck: 通用基线检测
                - tonalityImprove: 内容治理检测
                - aigcCheck: AIGC图片风险检测
                - aigcViolationDetection: AIGC图片侵权检测
                - aigcDetector: AIGC生成图片判断

        Returns:
            str | None: str=不通过原因, None=通过

        Author:
            zero
        """
        if not path:
            return None

        image_moderation_request = models.ImageModerationRequest(
            service=service,
            service_parameters=json.dumps({
                "imageUrl": path,
                "dataId": str(uuid.uuid1())
            })
        )

        runtime = util_models.RuntimeOptions()
        try:
            client = await cls.client()
            loop = asyncio.get_event_loop()
            func = partial(client.image_moderation_with_options, image_moderation_request, runtime)
            response = await loop.run_in_executor(None, func)
            
            if response.status_code == 200:
                result = response.body
                status = cls._handle_error(result.code)
                if status == "200":
                    reason = result.data.result[0]
                    if reason.risk_level != "low":
                        return f"{reason.description}({reason.risk_level})"
                    return None
                return status
            else:
                return None
        except Exception as err:
            return f"阿里内容审核异常: {str(err)}"

    @classmethod
    def _handle_error(cls, code: int):
        if code == 200:
            return "200"
        elif code == 400:
            return "可能是请求参数不正确导致，请仔细检查请求参数 (AliyunCensor)。"
        elif code == 408:
            return "可能是您的账号未授权、账号欠费、账号未开通、账号被禁等 (AliyunCensor)。"
        elif code == 500:
            return "可能是服务端临时出错。建议重试，若持续出现请联系阿里客服 (AliyunCensor)。"
        elif code == 581:
            return "可能是网络异常。建议重试，若持续出现请联系阿里客服 (AliyunCensor)。"
        elif code == 588:
            return "请求频率超出配额 (AliyunCensor)。"
        else:
            return "未知的错误 (AliyunCensor)。"
